package top.fullj.remoting.consumer;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.util.AttributeKey;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import top.fullj.remoting.codec.Hassian2Decoder;
import top.fullj.remoting.codec.Hassian2Encoder;

import java.net.URI;
import java.nio.ByteOrder;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author bruce.wu
 * @since 2022/2/15 9:46
 */
public class ConnectionPool {

    private static final Logger LOG = LoggerFactory.getLogger(ConnectionPool.class);

    private static final ConnectionPool INSTANCE = new ConnectionPool();

    public static ConnectionPool getInstance() {
        return INSTANCE;
    }

    static final AttributeKey<ServiceConfig> SERVER_CONFIG = AttributeKey.valueOf("SC");

    private final Map<String, Channel> pool = new ConcurrentHashMap<>();

    private final EventLoopGroup group = new NioEventLoopGroup();

    public Channel borrow(ServiceConfig config) {
        Channel channel;
        if (pool.containsKey(config.getUri())) {
            channel = pool.get(config.getUri());
            if (channel.isActive() && channel.isWritable()) {
                return channel;
            }
            //TODO release origin channel
        }
        channel = connect(config);
        channel.attr(SERVER_CONFIG).set(config);
        Channel origin = pool.putIfAbsent(config.getUri(), channel);
        if (origin != null) {
            //TODO release origin channel
        }
        return channel;
    }

    public void back(Channel channel) {
        //TODO check if need to release this channel
    }

    public void release() {
        for (Map.Entry<String, Channel> entry: pool.entrySet()) {
            entry.getValue().close();
        }
        group.shutdownGracefully();
    }

    private Channel connect(ServiceConfig config) {
        LOG.info("connecting to {}", config.getUri());
        URI uri = URI.create(config.getUri());
        try {
            return new Bootstrap()
                    .group(group)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.TCP_NODELAY, true)
                    .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
                    .handler(new ClientInitializer())
                    .connect(uri.getHost(), uri.getPort())
                    .sync()
                    .channel();
        } catch (InterruptedException e) {
            throw new RuntimeException("connecting to " + config.getUri() + " failed!", e);
        }
    }

    private static class ClientInitializer extends ChannelInitializer<SocketChannel> {
        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            ChannelPipeline pipeline = ch.pipeline();

            // binary codec
            pipeline.addLast(new LengthFieldBasedFrameDecoder(
                    ByteOrder.BIG_ENDIAN, Short.MAX_VALUE,
                    0, 2,
                    0, 2, true));
            pipeline.addLast(new LengthFieldPrepender(ByteOrder.BIG_ENDIAN,
                    2, 0, false));

            pipeline.addLast(new LoggingHandler(LogLevel.TRACE));

            pipeline.addLast(new Hassian2Decoder());
            pipeline.addLast(new Hassian2Encoder());

            pipeline.addLast(new ClientHandler());
        }
    }

}
